package com.test.spi;

import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.test.message.OnlineUserMessage;
import com.test.message.TextMessage;
import com.test.message.builder.MessageBuiler;
import com.test.message.wrapper.Payload;
import com.test.model.User;
import com.test.protocol.messages.AbstractMessage.QOSType;
import com.test.protocol.messages.ConnAckMessage;
import com.test.protocol.messages.ConnectMessage;
import com.test.protocol.messages.DisconnectMessage;
import com.test.protocol.messages.PubAckMessage;
import com.test.protocol.messages.PubCompMessage;
import com.test.protocol.messages.PubRecMessage;
import com.test.protocol.messages.PubRelMessage;
import com.test.protocol.messages.PublishMessage;
import com.test.protocol.messages.SubAckMessage;
import com.test.protocol.messages.SubscribeMessage;
import com.test.protocol.messages.UnsubscribeMessage;
import com.test.server.ServerChannel;
import com.test.server.netty.NettyChannel;
import com.test.topic.Topic;
import com.test.utils.SerializationUtil;

import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.EventExecutor;

public class ProtocolProcessorImpl implements IProtocolProcessor{
	
	private Map<String, ServerChannel> sessionStore = new ConcurrentHashMap<>();
	
	private Map<String, List<String>> topicStore = new ConcurrentHashMap<>();
	
	private static Logger LOGGER = LoggerFactory.getLogger(ProtocolProcessorImpl.class);

	@Override
	public void processConnect(ServerChannel session, ConnectMessage msg) {
		LOGGER.info("processConnect---------------------------------------");
		LOGGER.info("user connected, username:" + msg.getUsername() + " password:" + msg.getPassword());
		LOGGER.info(msg.toString());
		
		ConnAckMessage okResp = new ConnAckMessage();
		okResp.setQos(QOSType.MOST_ONE);
        okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
        session.write(okResp);
        
        session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, msg.getClientID());
        session.setAttribute(NettyChannel.ATTR_KEY_USERNAME, msg.getUsername());
        session.setAttribute(NettyChannel.ATTR_KEY_KEEPALIVE, msg.getKeepAlive());
        session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, msg.isCleanSession());
        sessionStore.put(msg.getClientID(), session);
        
        //群发在线列表
        for(String clientId : sessionStore.keySet()){
        	PublishMessage publishMessage = new PublishMessage();
            publishMessage.setQos(QOSType.MOST_ONE);
            publishMessage.setTopicName(Topic.CHAT);
            OnlineUserMessage userMessage = new OnlineUserMessage();
            for(String clientId1 : sessionStore.keySet()){
            	/*if(clientId1.equals(clientId)){
            		continue;
            	}*/
            	User user = new User();
            	user.setUserName(clientId1);
            	userMessage.getUsers().add(user);
            }
            publishMessage.setPayload(ByteBuffer.wrap(MessageBuiler.build(userMessage)));
        	ServerChannel clientSession = sessionStore.get(clientId);
        	clientSession.write(publishMessage);
        }
	}

	@Override
	public void processPubAck(ServerChannel session, PubAckMessage msg) {
		LOGGER.info("processPubAck---------------------------------------");
	}

	@Override
	public void processPublish(ServerChannel session, PublishMessage msg) {
		LOGGER.info("processPublish---------------------------------------");
		PubAckMessage pubAckMessage = new PubAckMessage();
        pubAckMessage.setMessageID(msg.getMessageID());
        session.write(pubAckMessage);
        
        Payload payload = MessageBuiler.getPayload(msg.getPayload().array());
        switch (payload.getType()) {
		case 1:
			TextMessage txMessage = SerializationUtil.deserialize(payload.getMessage(), TextMessage.class);
			int chatType = txMessage.getChatType();
			if(chatType == TextMessage.ONE_TO_GROUP){
				for(String clientId : sessionStore.keySet()){
		        	ServerChannel clientSession = sessionStore.get(clientId);
		        	PublishMessage publishMessage =  new PublishMessage();
		        	publishMessage.setPayload(msg.getPayload().duplicate());
		        	publishMessage.setTopicName(Topic.CHAT);
		        	publishMessage.setQos(QOSType.MOST_ONE);
		        	clientSession.write(publishMessage);
		        }
			}
			if(chatType == TextMessage.ONE_TO_ONE){
				ServerChannel clientSession = sessionStore.get(txMessage.getTo());
	        	PublishMessage publishMessage =  new PublishMessage();
	        	publishMessage.setPayload(msg.getPayload().duplicate());
	        	publishMessage.setTopicName(Topic.CHAT);
	        	publishMessage.setQos(QOSType.MOST_ONE);
	        	clientSession.write(publishMessage);
	        	
	        	session.write(msg);
			}
			break;
		case 2:
			
			break;
		default:
			break;
		}
	}

	@Override
	public void internalPublish(PublishMessage msg) {
		LOGGER.info("internalPublish---------------------------------------");
		
	}

	@Override
	public void processPubRel(ServerChannel session, PubRelMessage msg) {
		LOGGER.info("processPubRel---------------------------------------");
		
	}

	@Override
	public void processPubRec(ServerChannel session, PubRecMessage msg) {
		LOGGER.info("processPubRec---------------------------------------");
		
	}

	@Override
	public void processPubComp(ServerChannel session, PubCompMessage msg) {
		LOGGER.info("processPubComp---------------------------------------");
		
	}

	@Override
	public void processDisconnect(ServerChannel session) throws InterruptedException {
		LOGGER.info("processDisconnect---------------------------------------");
		String clientId = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
		sessionStore.remove(clientId);
		DisconnectMessage msg = new DisconnectMessage();
		msg.setMessageType(DisconnectMessage.DISCONNECT);
		msg.setQos(QOSType.MOST_ONE);
		session.write(msg);
	}

	@Override
	public void processConnectionLost(String clientID, boolean sessionStolen, NettyChannel channel) {
		LOGGER.info("processConnectionLost---------------------------------------");
		sessionStore.remove(clientID);
	}

	@Override
	public void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) {
		LOGGER.info("processUnsubscribe---------------------------------------");
		
	}

	@Override
	public void processSubscribe(ServerChannel session, SubscribeMessage msg) {
		LOGGER.info("processSubscribe---------------------------------------");
		SubAckMessage subAckMessage = new SubAckMessage();
		subAckMessage.setMessageID(msg.getMessageID());
		subAckMessage.addType(QOSType.EXACTLY_ONCE);
		subAckMessage.addType(QOSType.MOST_ONE);
		subAckMessage.addType(QOSType.LEAST_ONE);
		session.write(subAckMessage);
	}

}
